package com.movie.theatre.web.listener;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.common.message.MessageExt;
import org.springframework.stereotype.Component;
import org.springframework.util.CollectionUtils;

import java.util.List;

/**
 * @Auther: liwu
 * @Date: 2019/12/24 17:24
 * @Description:Order消息统一集结地
 */
@Slf4j
@Component
public class RouteMessageListenerConcurrently implements MessageListenerConcurrently {


    @Override
    public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) {
        String msgTopic = context.getMessageQueue().getTopic();
        if (CollectionUtils.isEmpty(msgs)) {
            log.warn("[收到消息]-[消息体为空], msgs={}, topic={}", msgs, msgTopic);
            return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
        }
        log.info("[收到消息]-[收到消息], size={}, topic={}, msgs={}", msgs.size(), msgTopic, msgs);
        try {
            for (MessageExt msg : msgs) {
                JSONObject json = JSONObject.parseObject(new String(msg.getBody()));
                log.info("[收到消息]-[收到消息....]" + JSON.toJSONString(json));

            }
            return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
        } catch (Throwable ex) {
            log.error("Message consumer error", ex);
            return ConsumeConcurrentlyStatus.RECONSUME_LATER;
        }
    }
}
